package com.xiaoke;

import java.io.IOException;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

import com.alibaba.fastjson.JSON;

public class ActorBestMoviesMain {
	public static class MovieMapper extends Mapper<LongWritable, Text, MovieInfo, NullWritable> {
		@Override
		protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
			MovieInfo movi = JSON.parseObject(value.toString(), MovieInfo.class);
			if (movi.getActor().indexOf("浅野忠信") != -1) {
				context.write(movi, NullWritable.get());
			} else {
				return;
			}
		}
	}

	public static class MovieReducer extends Reducer<MovieInfo, NullWritable, MovieInfo, NullWritable> {
		static int num = 0;

		@Override
		protected void reduce(MovieInfo key, Iterable<NullWritable> values, Context context)
				throws IOException, InterruptedException {

			for (NullWritable nullWritable : values) {
				if (num >= 5) {
					break;
				}
				num++;
				MovieInfo movieInfo = new MovieInfo();
				movieInfo.setTitle(key.getTitle());
				movieInfo.setType(key.getType());
				movieInfo.setActor(key.getActor());
				movieInfo.setStar(key.getStar());
				movieInfo.setYear(key.getYear());
				movieInfo.setFilm_page(key.getFilm_page());
				context.write(movieInfo, NullWritable.get());
			}
		}
	}

	public static void main(String[] args) throws IOException {
		// TODO 自动生成的方法存根
		// 1 获取配置信息，或者job对象实例
		Configuration configuration = new Configuration();
		Job job = Job.getInstance(configuration);

		// 2 指定本业务job要使用的mapper/Reducer业务类
		job.setJarByClass(ActorBestMoviesMain.class);
		job.setMapperClass(MovieMapper.class);
		job.setReducerClass(MovieReducer.class);
		job.addFileToClassPath(new Path("/lib/fastjson-1.2.62.jar"));

		// 3 指定mapper输出数据的kv类型
		job.setMapOutputKeyClass(MovieInfo.class);
		job.setMapOutputValueClass(NullWritable.class);

		// 4指定reducer的输出类型
		job.setOutputKeyClass(MovieInfo.class);
		job.setOutputValueClass(NullWritable.class);

		// 5 指定job的输入原始文件所在目录
		FileInputFormat.setInputPaths(job, new Path(args[0]));
		FileOutputFormat.setOutputPath(job, new Path(args[1]));

		try {
			job.waitForCompletion(true);
		} catch (ClassNotFoundException e) {
			// TODO 自动生成的 catch 块
			e.printStackTrace();
		} catch (InterruptedException e) {
			// TODO 自动生成的 catch 块
			e.printStackTrace();
		}

	}

}
